package com.zetyun.tiger.connectors.http.async;

import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.Method;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.AsyncAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;


@ActionMeta(
        id = "tiger.async-http-call",
        name = "Async Http Call",
        tags = {"async", "http"},
        category = "async",
        parameterClass = HttpCallerParam.class,
        description = "参考自 https://github.com/caarlos0-graveyard/flink-async-http-example "
)
@Slf4j
public class HttpCallerAction extends AsyncAction {

    private HttpCallerParam param;

    private Method method;

    private List<HttpParamPair> paramPairs;

    private JSONObject paramTemplate;

    @SneakyThrows
    @Override
    public void init(OperatorContext context) {
        super.init(context);
        param = context.getParameters();
        method = param.getRequestMethod();
        if (!method.equals(Method.POST)) {
            throw new IllegalArgumentException("暂不支持除POST之外的方法调用");
        }
        paramPairs = param.getParamPairs();
        paramTemplate = JSONUtil.parseObj(param.getParams());
    }

    @Override
    public CompletableFuture<Collection<RtEvent>> apply(RtEvent in) throws Exception {

        boolean needHttpCall = true;

        // assembly real args
        JSONObject argsObj = JSONUtil.createObj();
        for (HttpParamPair paramPair : paramPairs) {
            String argsName = paramPair.getArgsName();
            String argsValName = paramPair.getArgsValName();
            Object argsVal = in.getField(argsValName, "");
            if (StrUtil.isBlankIfStr(argsVal)) {
                needHttpCall = false;
            }
            argsObj.set(argsName, argsVal);
        }

        if (!needHttpCall) {
            in.setField(param.getRtHttpResField(), null);
            return CompletableFuture.completedFuture(Collections.singleton(in));
        } else {
            paramTemplate.putByPath(param.getArgsExpression(), argsObj);
            String argsStr = JSONUtil.toJsonStr(paramTemplate);

            return CompletableFuture.supplyAsync(
                    () -> {
                        JSONObject resObj;
                        if (method == Method.POST) {
                            resObj = JSONUtil.parseObj(
                                    HttpRequest
                                            .post(param.getRequestUrl())
                                            .timeout(param.getTimeoutMill())
                                            .body(argsStr)
                                            .execute(true)
                                            .body());
                        } else {
                            throw new IllegalArgumentException("暂不支持除POST之外的方法调用");
                        }
                        Boolean isSuccess = resObj.getByPath(param.getResFlagExpression(), Boolean.class);
                        if (isSuccess != null && isSuccess) {
                            JSONObject resDataObj = resObj.getByPath(param.getResDataExpression(), JSONObject.class);
                            in.setField(param.getRtHttpResField(), resDataObj);
                        } else {
                            in.setField(param.getRtHttpResField(), null);
                        }
                        return Collections.singleton(in);
                    });
        }


    }


    @Override
    public void close() throws Exception {
        param = null;
        paramTemplate = null;
        paramPairs = null;
        method = null;
    }
}
